/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.pre.parser.get;

import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.pre.parser.global.GlobalParas;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;
import net.sf.json.JSONObject;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.log4j.Logger;

/**
 *
 * @author moli
 */
public class ReadRecord {

    static Logger LOG = Logger.getLogger(ReadRecord.class);

    public void getRecord() {
        String fileName = null;
        BufferedReader fileReader = null;
        while (true) {
            try {
                fileName = GlobalParas.FILES_NAME_QUEUE.poll(2, TimeUnit.SECONDS);
            } catch (InterruptedException ex) {
                ExceptionUtils.getFullStackTrace(ex);
            }
            LOG.info("file queue num: "+GlobalParas.FILES_NAME_QUEUE.size());
            if(fileName == null){
                try {
                    Thread.sleep(5000);
                    continue;
                } catch (InterruptedException ex) {
                    LOG.error(ExceptionUtils.getFullStackTrace(ex));
                }
            }
            LOG.info(String.format("reading file: %s", GlobalParas.SOURCE_DIR + "/" + fileName));
            

            try {
                fileReader = new BufferedReader((new InputStreamReader(new FileInputStream(GlobalParas.SOURCE_DIR + "/" + fileName),"utf-8")));
//        (new FileReader(GlobalParas.SOURCE_DIR + "/" + fileName));
                String aRecord = null;
                while ((aRecord = fileReader.readLine()) != null) {
                    JSONObject jsonObject = JSONObject.fromObject(aRecord);
                    String msgType = jsonObject.getString("msg_type");
                    REMessage msg = GlobalParas.MSG_TYPE_MAP.get(msgType);
                    if (msg == null) {
                        LOG.error("configration does not include this msg, msgType is "+msgType);
                        continue;
                    }
                    List<Object> valueList = new ArrayList<>();
                    for (String key : msg.colNames) {
                        if (jsonObject.containsKey(key)) {
                            Object object = jsonObject.get(key);
//                            Object array = null;
//                            if (object instanceof Collection) {
//                                Collection c = (Collection) object;
//                                boolean first = true;
//
//                                int idx = 0;
//                                for (Object o : c) {
//                                    if (first) {
//                                        array = Array.newInstance(o.getClass(), c.size());
//                                        first = false;
//                                    }
//                                    Array.set(array, idx++, o);
//                                }
//                                valueList.add(array);
//                            } else {
                                valueList.add(object);
//                            }

                        } else {
                            valueList.add("");
                            LOG.error("key not found: " + key);
                        }
                    }
                    List<Object[]> datas = new ArrayList<Object[]>();
                    datas.add(valueList.toArray());
                    REMessage sendMsg = new REMessage(msg.colNames, msg.types, datas, msg.docSchemaName, msg.userDesc);

                    GlobalParas.SEND_QUEUE_MAP.get(msgType).put(sendMsg);

                }

            } catch (FileNotFoundException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            } catch (IOException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            } catch (InterruptedException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }
        }

    }
}
